package day1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

object RDD_2 {
  /*
    RDD算子分为 RDD、PairRDD、数值RDD
    RDD中的转换算子对于PairRDD、数值RDD都可以使用
   */
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir","D:\\hadoop")
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))
    Logger.getLogger("org").setLevel(Level.OFF)


    val arr =Array(1,2,3,4,5,6,7,8,9)
    val arr2 = Array("hello world","hello scala","hello spark")
    val arr5 = Array(("hello world","hello scala","hello spark"),2)

    //转换成算子
    val rdd1 = sc.parallelize(arr)
    val rdd2 = sc.parallelize(arr2)
    //RDD中的转换算子
    //map()算子:对于一个RDD做一对一遍历，将前一个RDD的元素读取出来处理完成之后导入到新的RDD中
    //前一个RDD和后一个RDD的长度相同
    val rdd1_map = rdd1.map(_ * 3)

    //RDD中的行动算子
    //count()算子:遍历出里面每个输出
    //foreach 算子:定义一个输出方式，,将对于RDD中的内存进行遍历输出
    //union 算子: 合并RDD，先将前一个RDD的元素赋值给当前RDD，再将后一个元素赋值给当前RDD
    //filter 算子:过滤、筛选、当前算子需要传递一个布尔值;当前值为true时将这个数据获取 会减少RDD长度
    //flat 算子:扁平化操作，参数、函数的返回值都为容器，容器返回的成员为当前RDD的元素
    //reduce 算子:
    //fold 算子:与reduce算子相类似，只是多了一个初始值


//    rdd1_map.foreach(println)
    val rdd1_union = rdd1.union(rdd1_map)
    //union
//    rdd1_union.foreach(println)

    //filter

    val rdd1_filter = rdd1.filter(i => if(i %2 ==0) true else false)

    val rdd1_filter2 = rdd1.filter(_ % 2 ==0)
//    rdd1_filter.foreach(println)

    //flatMap()
    val rdd1_flatMap = rdd1.flatMap(1 to _)

//    rdd1_flatMap.foreach(i => print(s"$i "))

    val rdd2_flatMap = rdd2.flatMap(_.split(" "))
//    rdd2_flatMap.foreach(println)
    //
    val rdd1_intersection =rdd1.intersection(rdd1_filter2)

//    rdd1_intersection.foreach(println)
    //并集
    val rdd2_intersection = rdd2.intersection(rdd2)

//    rdd2_intersection.foreach(println)

    val rdd1_subtract = rdd1.subtract(rdd1_filter)
    //差集
    rdd1_subtract.foreach(println)

    //cartesian() 算子: 笛卡尔积 将两个元素合并，前一个RDD作为Key 后一个RDD的每一个元素作为value 生成一个新的RDD
    val rdd1_cartesian = rdd1.cartesian(rdd2_flatMap)

//    rdd1_cartesian.foreach(println)

    //reduce() 算子：将RDD传入的数据聚合操作,类型相同,返回的内容与当前RDD同类型
    val rdd4 = sc.parallelize(Array(1,2,3,4,5,6,7,8,9),2)
//    rdd4.reduce((x,y) =>{
//      println(s"x=$x,y=$y")
//      x+y
//    })
//
//    println(rdd1.sum())
//
    //mapPartitions 算子L根据要求生成对应RDD，通过迭代器获得对于的数值
//    val rdd1_mapPart = rdd1.mapPartitions(iter => {
//      val list = iter.toList
//      val even = list.filter(i => i % 2 ==0).sum
//      val odd = list.filter(_ % 2 != 0).sum
//      Iterator(even,odd)
//    }
//    )
//      .foreach(println)

    val rdd2_mapPartIndex = rdd2.mapPartitionsWithIndex((index,iter) =>{
//      val list = iter.toList
      println(s"这是第$index 分区操作")
      iter.map(println(_))
    }
    ).foreach(println)

    //fold(a)(b)
    //a:初始值 b:分区内和分区间的操作
    println(rdd4.fold(0)(_+_))

    println(rdd4.fold(10)((x,y) =>{
      println(s"$x$y")
      x + y
    }
    ))
    //aggregate 算子:实现分区内和分区间不同的操作过程，并且返回的RDD类型与原RDD的类型可以不相同
    //拥有一个参数，其格式为aggregate(a)(b)(c)
    rdd4.aggregate((0,0))(
      (x,y) => (x._1 +y,x._2 +1),
      (x,y) => (x._1 +y._1,x._2 +y._2)
    )

  }
  }
